package com.reload.reaactor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Date: 2024/12/7 - 15:25
 * @Version: 17
 * @Author: Reloading
 * @Description:
 */
public class SinksDemo {
    public static void main(String[] args) throws Exception {
        utils();

        // 缓存
        cache();

        // 获得元素
        block();

        // 并发流
        parallelFlux();

        // 上下文切换
        context();
        // 阻塞主线程
        System.in.read();
    }


    /**
     * ThreadLocal在响应式编程中无法使用
     * 响应式中数据流期间共享数据使用 Context API
     * Context 读写
     * ContextView 只读
     * ThreadLocal共享数据,上游的的所有人都能看到
     * Context有下游传播给上游,上游能拿到下游的最近一次数据
     */
    private static void context() {
        Flux.just(1, 2, 3)
                .transformDeferredContextual((data, context) -> {
                    System.out.println("data = " + data);
                    System.out.println("context = " + context);
                    return data.map(i -> i + "==>" + context.get("prefix"));
                })
                // ThreadLocal共享了数据，上游的所有人能看到
                // Context由下游传播给上游
                .contextWrite(Context.of("prefix", "哈哈"))
                .subscribe(v -> System.out.println("v = " + v));
    }

    private static void parallelFlux() {
        Flux.range(1, 1000)
                // 10个一组
                .buffer(10)
                // 8个线程并行
                .parallel(8)
                .runOn(Schedulers.newParallel("yy"))
                .log()
                .flatMap(Flux::fromIterable)
                .subscribe(v -> System.out.println("v = " + v));
    }

    private static void block() {
        Integer integer = Flux.just(1, 2, 3)
                .map(i -> i + 10)
                .blockLast();
        System.out.println("integer = " + integer);

        List<Integer> block = Flux.just(1, 2, 3, 4)
                .map(i -> i + 10)
                .collectList()
                .block();
        System.out.println("block = " + block);
    }

    private static void cache() {
        Flux<Integer> cache = Flux.range(1, 10)
                .delayElements(Duration.ofSeconds(1))
                // 缓存N个数据  默认全部缓存
                .cache(1);
        cache.subscribe();

        new Thread(() -> {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            cache.subscribe(v -> System.out.println("v = " + v));
        }).start();
    }

    private static void utils() {
        // 发送Flux数据
        Sinks.many();
        // 发送Mono数据
        Sinks.one();

        // 单播 这个管道只能绑定单个订阅者
        // SinksDemo.many().unicast();
        // 多播 可以绑定多个消费者
        // SinksDemo.many().multicast();
        // 重播 能重放元素。是否给后来的订阅者把之前的元素依旧发给他
        // SinksDemo.many().replay();
        Sinks.Many<Object> many = Sinks.many()
                .unicast()
                // 背压队列
                .onBackpressureBuffer(new LinkedBlockingQueue<>(5));

        // 默认订阅者从订阅的那一刻开始接收数据
        // 设置重放功能，底层利用队列进行缓存数据
        Sinks.Many<Object> limit = Sinks.many().replay().limit(3);
        // 发数据
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                many.tryEmitNext("a-" + i);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }).start();
        // 订阅 many.asFlux()转为Flux流
        many.asFlux().subscribe(v -> System.out.println("v = " + v));
    }
}
